package com.ds.infrastructure.hyperspace.container.internal.gzs;

import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.zonesdk.ZoneClient;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author  Weishaoying
 */
public class GzsChangeRegistry {

    private static final String DELIMITER = ":";

    private Map<String, GzsChangeContextListener> gzsChangeListenerMap =
            new ConcurrentHashMap<>();

    private final ZoneClient gzsClient;

    public GzsChangeRegistry(ZoneClient gzsClient) {
        this.gzsClient = gzsClient;
    }

    public void register(HyperspaceDuplexFlowContext duplexFlowContext) {
        GzsChangeContextListener listener = gzsChangeListenerMap.computeIfAbsent(
                getGzsChangeMapKey(duplexFlowContext), k -> {
                    GzsChangeContextListener tempListener = new GzsChangeContextListener();
                    this.gzsClient.setListener(tempListener.getSdkZoneChangeListener(),
                            duplexFlowContext.getShardingKeyType(), duplexFlowContext.getShardingKey());
                    return tempListener;
                });
        listener.addContext(duplexFlowContext);
    }

    public void unRegister(HyperspaceDuplexFlowContext duplexFlowContext) {
        gzsChangeListenerMap.computeIfPresent(getGzsChangeMapKey(duplexFlowContext), (key, listener) -> {
            listener.removeContext(duplexFlowContext);
            if (listener.isEmpty()) {
                this.gzsClient.removeListener(duplexFlowContext.getShardingKeyType(),
                        duplexFlowContext.getShardingKey());
                return null;
            }
            return listener;
        });
    }

    private String getGzsChangeMapKey(HyperspaceDuplexFlowContext duplexFlowContext) {
        return new StringBuilder().append(duplexFlowContext.getShardingKeyType())
                .append(DELIMITER).append(duplexFlowContext.getShardingKey()).toString();
    }

}
